package com.qupeng.demo.kafka.kafkaspringbootconsumer.listener.single;

import com.qupeng.demo.kafka.kafkaspringbootconsumer.model.Product;
import com.qupeng.demo.kafka.kafkaspringbootconsumer.config.KafkaProperties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@ConditionalOnProperty(name = "app.kafka.consumer.mode.single-listener", havingValue = "true", matchIfMissing = false)
@Component
public class KafkaSingleListener {

    @Autowired
    private KafkaProperties kafkaProperties;

    @KafkaListener(id = "single-listener",
            groupId = "#{kafkaProperties.getGroupId()}",
            topics = "#{kafkaProperties.getTopics()}",
            clientIdPrefix = "single-listener",
            properties = {}
    )
    public void listen(ConsumerRecord<Integer, Product> consumerRecord) {
        System.out.println(consumerRecord);
        System.out.println(consumerRecord.value());
    }
}
